package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

public class DwdTradePayDetailSuc {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //todo 设置状态的ttl  生产环境设置为最大乱序程度
        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(905));

        //todo 2.获取topicdb数据并过滤成功支付的数据
        tenv.executeSql(MyKafkaUtil.getTopicDb("pay_detail_suc"));


        Table paymentInfoTable = tenv.sqlQuery("select " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                "`pt`  " +
                "from topic_db " +
                "where `table` = 'payment_info' " +
                "and `type` = 'update' " +
                "and data['payment_status']='1602' ");

        tenv.createTemporaryView("payment_info",paymentInfoTable);

//        tenv.sqlQuery("select * from payment_info").execute().print();

        //todo 3.消费下单主题数据

        tenv.executeSql("create table dwd_trade_order_detail ( " +
                "id string, " +
                "order_id string, " +
                "sku_id string, " +
                "sku_name string, " +
                "order_price string, " +
                "sku_num string, " +
                "create_time string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "source_id string, " +
                "split_total_amount string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "user_id string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string, " +
                "row_op_ts timestamp_ltz(3) " +
                ") "+MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail","pay_detail_suc_order"));


        //todo 4.读取mysql base_dic表

        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());


        //todo 5.三表关联
        Table resTable = tenv.sqlQuery(
                "select " +
                        "od.id order_detail_id, " +
                        "od.order_id, " +
                        "od.user_id, " +
                        "od.sku_id, " +
                        "od.sku_name, " +
                        "od.province_id, " +
                        "od.activity_id, " +
                        "od.activity_rule_id, " +
                        "od.coupon_id, " +
                        "pi.payment_type payment_type_code, " +
                        "dic.dic_name payment_type_name, " +
                        "pi.callback_time, " +
                        "od.source_id, " +
                        "od.source_type_id, " +
                        "od.source_type_name, " +
                        "od.sku_num, " +
                        "od.order_price, " +
                        "od.split_activity_amount, " +
                        "od.split_coupon_amount, " +
                        "od.split_total_amount split_payment_amount, " +
                        "od.row_op_ts row_op_ts " +
                        "from payment_info pi " +
                        "join dwd_trade_order_detail od " +
                        "on pi.order_id = od.order_id " +
                        "join `base_dic` for system_time as of pi.pt as dic " +
                        "on pi.payment_type = dic.dic_code");

        tenv.createTemporaryView("res_table",resTable);

        tenv.toAppendStream(resTable, Row.class).print();


        //TODO 6.创建Kafka支付成功表


        tenv.executeSql("create table dwd_trade_pay_detail_suc (" +
                "order_detail_id string, " +
                "order_id string, " +
                "user_id string, " +
                "sku_id string, " +
                "sku_name string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string, " +
                "payment_type_code string, " +
                "payment_type_name string, " +
                "callback_time string, " +
                "source_id string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "sku_num string, " +
                "order_price string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "split_payment_amount string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(order_detail_id) not enforced " +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));


        //todo 将数据写出
        tenv.executeSql("insert into dwd_trade_pay_detail_suc select * from res_table");




        env.execute();
    }
}
